package client

import (
	"github.com/gobwas/ws/wsutil"
	"github.com/panjf2000/gnet/v2"
	"github.com/xtaci/kcp-go/v5"
	"go.uber.org/atomic"
	"net"
	"sync"
	"wths/wthsjy-im/plugins/log"
	"wths/wthsjy-im/protocol"
	"wths/wthsjy-im/server/handle"
)

type Client struct {
	once           sync.Once
	authed         atomic.Bool
	closed         atomic.Bool
	clientId       string
	conn           net.Conn
	messageHandler handle.MessageHandle
	topicHandler   handle.TopicHandle
	topics         sync.Map
	IsWebsocket    bool
}

func NewClient(conn net.Conn, messageHandler handle.MessageHandle, topicHandler handle.TopicHandle) *Client {
	c := &Client{
		conn:           conn,
		messageHandler: messageHandler,
		topicHandler:   topicHandler,
	}
	return c
}

func (c *Client) ClientID() string {
	return c.clientId
}

func (c *Client) HandleProtocol(p *protocol.Packet) {
	if p == nil {
		return
	}
	log.Debug("HandleProtocol").Msg(p.Head(), p.Data(), string(p.Data()))
	// 认证
	if p.Head() == protocol.CONNECT {
		if err := c.connect(p); err != nil {
			log.Error("handleProtocol").Msg(err)
			c.Close()
		}
		return
	}

	log.Debug("Handle").Msg(c.ClientID())

	// 没有认证，关闭
	if !c.authed.Load() {
		c.Close()
		log.Debug("Handle").Msg("un authed", c.ClientID())
		return
	}
	if c.Closed() {
		log.Debug("Handle").Msg("closed", c.ClientID())
		return
	}

	switch p.Head() {
	case protocol.PUBLISH:
		if err := c.publish(p); err != nil {
			c.Close()
		}
	case protocol.DISCONNECT:
		if err := c.disconnect(p); err != nil {
			c.Close()
		}
	case protocol.PING:
		if err := c.ping(p); err != nil {
			c.Close()
		}
	case protocol.Subscribe:
		if err := c.subscribe(p); err != nil {
			c.Close()
		}
		// 处理并发，如果先close 然后再执行的订阅，这里需要回滚
		if c.Closed() {
			log.Debug("###@Handle").Msg("closed 回滚", c.ClientID())
			c.removeAllBindTopic()
		}
	case protocol.UnSubscribe:
		if err := c.unsubscribe(p); err != nil {
			c.Close()
		}
	}
}

// 取消订阅
func (c *Client) unsubscribe(p *protocol.Packet) error {
	info, err := protocol.DecodeUnsubscribe(p.Data())
	if err != nil {
		return err
	}
	acks := c.messageHandler.UnSubscribe(c.clientId, info)
	for topicName, ok := range acks {
		if ok {
			c.topicHandler.Unbind(topicName, c)
			c.topics.Delete(topicName)
		}
	}
	pkg, err := protocol.NewSubAckPacket(acks)
	if err != nil {
		return err
	}
	b, err := protocol.Encode(pkg)
	if err != nil {
		return err
	}
	_ = c.Write(b)
	return nil
}

// 订阅
func (c *Client) subscribe(p *protocol.Packet) error {
	log.Debug("###subscribe").Msg(c.ClientID())
	info, err := protocol.DecodeSubscribe(p.Data())
	if err != nil {
		log.Error("###subscribe").Msg(c.ClientID(), err)
		return err
	}

	acks := c.messageHandler.Subscribe(c.clientId, info)
	for topicName, ok := range acks {
		if ok {
			c.topics.Store(topicName, struct{}{})
			c.topicHandler.Bind(topicName, c)
		}
	}

	pkg, err := protocol.NewSubAckPacket(acks)
	if err != nil {
		log.Error("###subscribe").Msg(c.ClientID(), err)
		return err
	}
	b, err := protocol.Encode(pkg)
	if err != nil {
		log.Error("###subscribe").Msg(c.ClientID(), err)
		return err
	}
	_ = c.Write(b)
	return nil
}

// 心跳
func (c *Client) ping(p *protocol.Packet) error {
	info, err := protocol.DecodePing(p.Data())
	if err != nil {
		return err
	}
	c.messageHandler.Ping(c.clientId, info)
	return nil
}

// 离线
func (c *Client) disconnect(p *protocol.Packet) error {
	info, err := protocol.DecodeDisConnect(p.Data())
	if err != nil {
		return err
	}
	c.messageHandler.Disconnect(c.clientId, info)
	return nil
}

// 发布消息
func (c *Client) publish(p *protocol.Packet) error {
	info, err := protocol.DecodePublish(p.Data())
	if err != nil {
		return err
	}
	c.messageHandler.Publish(c.clientId, info)
	return nil
}

// 认证
func (c *Client) connect(p *protocol.Packet) error {
	if c.authed.Load() {
		return nil
	}
	info, err := protocol.DecodeConnect(p.Data())
	if err != nil {
		return err
	}
	clientId, pass := c.messageHandler.Connect(info)
	if !pass {
		return protocol.ErrAuth
	}

	c.clientId = clientId
	c.authed.Store(true)

	ackPacket, _ := protocol.NewConnectAckPacket(true, "login success")
	b, _ := protocol.Encode(ackPacket)
	_ = c.Write(b)

	return nil
}

func (c *Client) Close() {
	c.once.Do(func() {
		log.Debug("Close2").Msg(c.ClientID())
		c.closed.Store(true)
		//移除订阅关系
		c.removeAllBindTopic()
		_ = c.conn.Close()
	})
}

func (c *Client) Closed() bool {
	return c.closed.Load()
}

func (c *Client) Write(b []byte) (err error) {
	switch c.conn.(type) {
	case gnet.Conn:
		if c.IsWebsocket {
			err = wsutil.WriteServerBinary(c.conn, b)
		} else {
			err = (c.conn).(gnet.Conn).AsyncWrite(b, nil)
		}
	case *kcp.UDPSession:
		_, err = c.conn.(*kcp.UDPSession).Write(b)
	}
	if err != nil {
		log.Error("Write").Msg(err)
	}
	return err
}

func (c *Client) removeAllBindTopic() {
	c.topics.Range(func(topicName, value any) bool {
		c.topicHandler.Unbind(topicName.(string), c)
		return true
	})
}
